package club.monkeywood.ad.dmp.tag

import club.monkeywood.ad.dmp.util.{JedisPools, TagsUtils}
import com.typesafe.config.ConfigFactory
import org.apache.hadoop.hbase.client.{ConnectionFactory, Put}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import scala.collection.mutable.ListBuffer


object Tags4CtxTest extends App {

	if (args.length != 5) {
		println(
			"""
				|cn.dmp.tags.Tags4Ctx
				|参数：
				| 日志输入路径
				| APP字典文件路径
				| 停用词库
				| 日期
				| 输出路径
			""".stripMargin)
		sys.exit()
	}

	val Array(inputPath, dictFilePath, stopWordsFilePath, day, outputPath) = args

	// 2 创建sparkconf->SparkSession
	val sparkConf = new SparkConf()
	sparkConf.setAppName(s"${this.getClass.getSimpleName}")
	sparkConf.setMaster("local[*]")
	//使用KryoSerializer更快
	// RDD 序列化到磁盘 worker与worker之间的数据传输
	sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
	//避免字段名过长报错
	sparkConf.set("spark.debug.maxToStringFields", "100")
	val ss = SparkSession.builder()
		.appName(s"${this.getClass.getSimpleName}")
		.config(sparkConf)
		.getOrCreate()

	val df: DataFrame = ss.read.parquet(inputPath)
	df.createTempView("log")
	df.show()

	// 读取日志parquet文件
	val logRdd: RDD[Row] = ss.read.parquet(inputPath)
		.where(TagsUtils.hasSomeUserIdConditition) //过滤掉没有用户id的数据
		.rdd

//	val tags: Array[(String, List[(String, Int)])] = logRdd.map(row => {
//		// 行数据进行标签化处理
//		// 广告标签
//		val ads: Map[String, Int] = Tags4Ads.makeTags(row)
//
//		//设备标签
//		val devices: Map[String, Int] = Tags4Devices.makeTags(row)
//
//		//用户id标签
//		//返回的来源于多种渠道的用户id
//		val allUserId: ListBuffer[String] = TagsUtils.getAllUserId(row)
//
//		//用户id暂时未用，取第一个id占位即可
//		val userID: String = allUserId(0)
//		val list: List[(String, Int)] = (ads ++ devices).toList
//		(userID, list)
//	}).collect()

	val tags: Array[(String, List[(String, Int)])] = logRdd.mapPartitions(par => {

		//val listBuffer = new collection.mutable.ListBuffer[(String, List[(String, Int)])]()

		//注意：不能用map，否则输出为空
		val tagsOfPar: Iterator[(String, List[(String, Int)])] = par.map(row => {

			// 行数据进行标签化处理
			// 广告标签
			val ads: Map[String, Int] = Tags4Ads.makeTags(row)

			//设备标签
			val devices: Map[String, Int] = Tags4Devices.makeTags(row)

			//用户id标签
			//返回的来源于多种渠道的用户id
			val allUserId: ListBuffer[String] = TagsUtils.getAllUserId(row)

			//用户id暂时未用，取第一个id占位即可
			val userID: String = allUserId(0)
			val list: List[(String, Int)] = (ads ++ devices).toList
			//listBuffer.append((userID, list))
			(userID, list)
		})

		tagsOfPar

		//listBuffer.iterator

	}).collect()

	println(tags.toString)

	ss.close()

}
